package com.etl.dataflow.executor.communication;

import cn.hutool.http.HttpResponse;
import cn.hutool.http.HttpUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.etl.dataflow.common.entity.ExecutorInfo;
import com.etl.dataflow.common.exception.EtlException;
import com.etl.dataflow.common.response.Code;
import com.etl.dataflow.common.util.IpUtil;
import com.etl.dataflow.executor.util.SchedulerInfoUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

/**
 * @author dx
 * @since 2022/4/20
 */
@Slf4j
@Component("schedulerInterface")
public class SchedulerInterface {

    private static final String CODE = "code";
    @Resource
    private SchedulerInfoUtil schedulerInfoUtil;
    @Value("${server.port}")
    private String executorPort;
    
    public void register() {
        boolean success = false;
        while (!success) {
            try {
                tryRegister();
                success = true;
            } catch (EtlException e) {
                e.printStackTrace();
                // 每5s尝试重连
                try {
                    TimeUnit.SECONDS.sleep(5);
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                }
            }
        }
    }

    public void tryRegister() {
        String body;
        try (HttpResponse response = HttpUtil.createPost(this.getRegisterUrl())
                .body(JSONUtil.toJsonStr(this.getCurrentExecutorInfo()), "application/json")
                .execute()) {
            body = response.body();
        }
        if (JSONUtil.isJsonObj(body)) {
            JSONObject json = JSONUtil.parseObj(body);
            if (Code.SUCCESS.getCode() == (Integer) json.get(CODE)) {
                log.debug("注册到任务调度中心成功。");
                return;
            }
            throw new EtlException("无法注册到任务调度中心：" + body);
        }
        throw new EtlException("无法注册到任务调度中心：" + body);
    }

    private ExecutorInfo getCurrentExecutorInfo() {
        ExecutorInfo executorInfo = new ExecutorInfo();
        executorInfo.setHost(IpUtil.getIpAddress());
        executorInfo.setPort(this.executorPort);
        executorInfo.setUuid(ExecutorMark.getInstance().getUuid());
        return executorInfo;
    }

    private String getRegisterUrl() {
        return "http://" + schedulerInfoUtil.getSchedulerUrl() + "/register";
    }

}
